/* 
 * Quechua - the lightweight data mining framework
 *
 * Copyright (C) 2012 Marek Denis <quechua@octogan.net>
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

#include "../include/workflow.h"
#include "../include/quechua.h"

Workflow::Workflow(const string_t& name_,ev::default_loop *loop) : name(name_),
step(IDLE), type(TRIGGERED) {
    init(loop);
}

Workflow::~Workflow() {
    clear_queue_t(input_queue);
    clear_queue_t(alg_queue);
    clear_queue_t(postproc_queue);
    clear_queue_t(logger_queue);

    functions.channel  = NULL;
    functions.logger   = NULL;

    delete functions.processor;
    delete functions.algorithm;

    LOG(INFO) << "Workflow " << name << " destroyed";
}

STEP Workflow::get_next_step(STEP step) {
    STEP step_;
    switch(step) {
        case IDLE:
            step_ = PREPROCESSING;
            break;
        case PREPROCESSING:
            step_ = PREPROCESSED;
            break;
        case PREPROCESSED:
            step_ = ALGRUNNING;
            break;
        case ALGRUNNING:
            step_ = ALGDONE;
            break;
        case ALGDONE:
            step_ = POSTPROCESSING;
            break;
        case POSTPROCESSING:
            step_ = POSTPROCESSED;
            break;
        case POSTPROCESSED:
            step_ = LOGGING;
            break;
        case LOGGING:
            step_ = DONE;
            break;
        case DONE:
            step_ = IDLE;
            break;
        default:
            step_ = IDLE;
            break;
    };
    return step_;
};

STEP Workflow::next_step() {
    step = get_next_step(step);
    return step;
}

const string_t& Workflow::getname() const {
    return name;
};


void Workflow::setup_scheduler(const setting_t& flow_config) {
    if(type==SCHEDULED) {
        string_t schedule = "1H";
        ev_tstamp seconds = 3600.;
        if(flow_config.exists("schedule"))
            flow_config.lookupValue("schedule",schedule);
        seconds =(ev_tstamp) Python::convert_to_seconds(schedule); 
        periodic_watcher.set<Workflow,&Workflow::schedule_cb>(this);
        periodic_watcher.set(0.,seconds);
        periodic_watcher.set(*g_quechua->loop);
        LOG(INFO) << "Workflow " << name << " scheduled for "
                  << (double)seconds << " seconds interval";
    } 
};

void Workflow::setup_type(const setting_t& flow_config) {
    if(flow_config.exists("type")) {
        string_t w_type;
        flow_config.lookupValue("type",w_type);
        if(w_type.compare("scheduled")==0) {
            this->type = SCHEDULED;
            LOG(INFO) << "Workflow " << name << " working type is SCHEDULED";
        }
        else if(w_type.compare("triggered")==0) {
            this->type = TRIGGERED;
            LOG(INFO) << "Workflow " << name << " working type is TRIGGERED";
        }
        else if(w_type.compare("oneshot")==0) {
            this->type = ONESHOT;
            LOG(INFO) << "Workflow " << name << " working type is ONESHOT";
        }
        else {
            this->type = TRIGGERED;
            LOG(WARN) << "Workflow " << name << " type is undefined, setting to TRIGGERED";
        }
    } 
    else {
        this->type = TRIGGERED;
        LOG(INFO) << "Workflow " << name << " default working type is TRIGGERED";
    }
};

void Workflow::iterate_cb(ev::async &watcher, int revents) {
    iterate();
};

void Workflow::iterate() {
#ifdef DEBUG_MSG
    {LOG(DEBUG) << "Workflow " << name << "  iterate_cb() start";}
#endif
    bool repeat;
    do {
        repeat = false;
        datapack_ptr data;
        data.reset();
        switch(step) {
           case IDLE:
               if(input_queue.empty())
                   return;
               next_step();
               data = input_queue.front();
               input_queue.pop();
               //check if the preprocessor exists
               if(functions.processor==NULL) {
#ifdef DEBUG_MSG
                   LOG(DEBUG) << "Processor not set, "
                              << "step now is PREPROCESSED";
#endif

                  alg_queue.push(data);
                  step = PREPROCESSED;

                  repeat = true;
               }
               else {
                  g_quechua->threads->RUN_IN_THREAD(data,
                          (void*)functions.processor,this,
                          &serve_processor);
               }
               data.reset();
#ifdef DEBUG_QUEUES
               debug_queue_t_size("input_queue",input_queue);
#endif
               break;
           case PREPROCESSING:
               break;
           case PREPROCESSED:
               if(alg_queue.empty())
                   return;

               data = alg_queue.front();
               alg_queue.pop();

               if(functions.algorithm==NULL) {
#ifdef DEBUG_MSG
                   LOG(DEBUG) << "Algorithm not set, "
                              << "setting step to ALGDONE";
#endif
                   postproc_queue.push(data);
                   step = ALGDONE;
                   repeat = true;
               }
               else {
                  next_step();
                  g_quechua->threads->RUN_IN_THREAD(data,
                       (void*)functions.algorithm,this,
                       &serve_algorithm);
               }
               data.reset();
#ifdef DEBUG_QUEUES
               debug_queue_t_size("alg_queue",alg_queue);
               assert(alg_queue.empty());
#endif

               break;
           case ALGRUNNING:
               break;
           case ALGDONE:
               if(postproc_queue.empty())
                   return;
               next_step();
               data = postproc_queue.front();
               postproc_queue.pop();
               if(functions.processor==NULL) {
#ifdef DEBUG_MSG
                  LOG(DEBUG) << "Processor not set, "
                             << "setting step to POSTPROCESSED";
#endif
                  logger_queue.push(data);
                  step = POSTPROCESSED;
                  repeat = true;
               }
               else {
                   g_quechua->threads->RUN_IN_THREAD(data,
                          (void*)functions.processor,this,
                          &serve_processor);
               }
               data.reset();
#ifdef DEBUG_QUEUES
               debug_queue_t_size("postproc_queue",postproc_queue);
               assert(postproc_queue.empty());
#endif

               break;
           case POSTPROCESSING:
               break;
           case POSTPROCESSED:
               if(logger_queue.empty())
                   return;
               next_step();
               data = logger_queue.front();
               logger_queue.pop();
               if(functions.logger==NULL) {
#ifdef DEBUG_MSG
                  LOG(DEBUG) << "Logger not set,"
                             << "discarding the data "
                             << "setting step to DONE";
#endif
                  step = DONE;
                  repeat = true;
                  data.reset();
                  break;
               }
               functions.logger->logwork(data);
               data.reset();
#ifdef DEBUG_QUEUES
               debug_queue_t_size("logger_queue",logger_queue);
               assert(logger_queue.empty());
#endif
               next_step();
               repeat = 1;
               break;
           case LOGGING:
               break;
           case DONE:
               check_working_type();
               step = IDLE;
               repeat = true;
#ifdef DEBUG_QUEUES
               debug_queue_t_size("input_queue",input_queue);
#endif
               if(!input_queue.empty()) repeat = true;
               cyclic_reset();
               break;
            case STOPPED:
               repeat = false;
            default:
               break;
        }; //switch
    } while(repeat);
#ifdef DEBUG_MSG
    {LOG(DEBUG) << "Workflow " << name << " iterate_cb() stop";}
#endif
}

void Workflow::schedule_cb(ev::periodic &watcher, int revents) {
    if(functions.channel==NULL) {
        LOG(ERROR) << "Trying to notify channel whereas channel is not available";
        return;
    }
    LOG(DEBUG) << "Workflow " << name << " notyfing channel " << functions.channel->internal->name;
    functions.channel->notify();
};

void Workflow::schedule_iterate_cb(ev::periodic &watcher, int revents) {
    iterate();
};

void Workflow::init(ev::default_loop *loop) {
    functions.channel = NULL;
    functions.processor = NULL;
    functions.algorithm = NULL;
    functions.logger = NULL;

    async_watcher.set<Workflow,&Workflow::iterate_cb>(this);
    async_watcher.set(*loop);

    iterate_watcher.set<Workflow,&Workflow::schedule_iterate_cb>(this);
    iterate_watcher.set(0.,60.);
    iterate_watcher.set(*loop);
};

void Workflow::start() {
    async_watcher.start();
    iterate_watcher.start();
    if(type==SCHEDULED) {
        periodic_watcher.start();
        functions.channel->notify();
    }
    else if(functions.channel && type==ONESHOT) {
        LOG(DEBUG) << "Notifying channel";
        functions.channel->notify();
    }
};   

void Workflow::stop() {
    async_watcher.stop();                                                                                                
    if(type==SCHEDULED)
        periodic_watcher.stop();
    iterate_watcher.stop();
    step = STOPPED;
};

// load algorithm and processor and init them
void Workflow::init_internals(const setting_t& flow_config) {
    
    if(flow_config.exists("algorithm")) {
         setting_t& algorithm_settings = flow_config["algorithm"];
         string_t algorithm_name;
         
         if(algorithm_settings.lookupValue("name",algorithm_name)) {

             Algorithm* algorithm;

             if((algorithm = g_quechua->modules->load_algorithm(algorithm_name,flow_config["algorithm"]))==NULL) {
                 LOG(ERROR) << "Algorithm " << algorithm_name << " wasn't loaded" ;        
             }
             else 
                 LOG(INFO) << "Algorithm " << algorithm->internal->name << " loaded";
             
             functions.algorithm = algorithm;
         } else {
             LOG(WARN) << "Cannot find algorithm name in the configuration file, skipping";
             functions.algorithm = NULL;
         }
    } else {
        LOG(WARN) << "No algorithm section...";
    }


   if(flow_config.exists("processor")) { 
    setting_t& processor_settings = flow_config["processor"];
    string_t processor_name;
    if(processor_settings.lookupValue("name",processor_name)) {

        Processor* processor;
        
        if( (processor = g_quechua->modules->load_processor(processor_name,flow_config["processor"]))==NULL) {
            LOG(ERROR) << "Processor " << processor_name << " wasn't loaded" ;
        }
        else {
            LOG(INFO) << "Processor " << processor->internal->name << " loaded";
        }
        functions.processor = processor;
    } else {
        LOG(WARN) << "Cannot find processor name in the configuration file, skipping";
        functions.processor = NULL;
    }
   } else {
       LOG(WARN) << "No processor section...";
   }

    setup_type(flow_config);
    setup_scheduler(flow_config);

    return;
};

bool Workflow::add_channel(Channel* channel) {
    if(functions.channel!=NULL) {
        LOG(WARN) << "Channel already added, delete it and try again";
        return false;
    }
    LOG(INFO) << "Adding channel " << channel->internal->name;
    functions.channel = channel;
    return true;
};

bool Workflow::add_logger(Logger* logger) {
    if(functions.logger!=NULL) {
        LOG(WARN) << "Logger already added, delete it and try again";
        return false;
    }
    functions.logger = logger;
    return true;
};


void Workflow::rem_channel() {
    functions.channel = NULL;
    return;
}

void Workflow::rem_logger() {
    functions.logger = NULL;
    return;
}

void Workflow::cyclic_reset() {
    if(functions.processor) functions.processor->reset();
    if(functions.algorithm) functions.algorithm->reset();
}

void Workflow::notify(datapack_ptr ptr, STEP step_) {
#ifdef DEBUG_MSG
    LOG(DEBUG) << "Workflow " << name << " notify() start";
#endif
    if(step_==this->step || step_ == IDLE) {
      switch(step_) {
          case IDLE:
              input_queue.push(ptr);
              break;
          case PREPROCESSING:
              alg_queue.push(ptr);
              break;
          case ALGRUNNING:
              postproc_queue.push(ptr);
              break;
          case POSTPROCESSING:
              logger_queue.push(ptr);
              break;
          default:
              return;
      };
    }

    if(step_!=IDLE) next_step();
    async_watcher.send(); 

#ifdef DEBUG_MSG
    LOG(DEBUG) << "Workflow::notify stop";
#endif
};

void Workflow::check_working_type() {
    if(this->type==ONESHOT) {
        LOG(INFO) << "Working type is set to "
                  << " oneshot...work done, exiting";
        g_quechua->loop->unloop();
    }
};

void clear_queue_t(queue_t& q) {
    while(!q.empty()) {
        q.pop();
    }
};

